--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
Commit cae7f29d2d14b96d345e903a90f5b02aa171e617
Parents : 70cf18a
Author : Mark Qvist <bc7291552be7a58f361522990465165c>
Signature : T66BB85Valid, signed by author
Date : 2026-07-26T02:38:06+02:00
Added fixed locations support
Changes
2 files changed, 63 insertions(+), 8 deletions(-)
Diff
diff --git a/sbapp/sideband/core.py b/sbapp/sideband/core.py
index 6268e157..f84de141 100644
--- a/sbapp/sideband/core.py
+++ b/sbapp/sideband/core.py
@@ -168,6 +168,7 @@ class SidebandCore():
self.telemetry_changes = 0
self.telemetry_response_excluded = []
self.telemetry_interfaces = {}
+ self.fixed_locations = {}
self.pending_telemetry_send = False
self.pending_telemetry_send_try = 0
self.pending_telemetry_send_maxtries = 2
@@ -361,8 +362,7 @@ class SidebandCore():
except Exception as e:
RNS.log("Error while configuring Reticulum instance: "+str(e), RNS.LOG_ERROR)
- else:
- pass
+ else: pass
self.active_propagation_node = None
self.propagation_detector = PropagationNodeDetector(self)
@@ -371,6 +371,11 @@ class SidebandCore():
RNS.Transport.register_announce_handler(self)
RNS.Transport.register_announce_handler(self.propagation_detector)
+ def load_fixed_job():
+ time.sleep(1)
+ self.__load_fixed_locations()
+ threading.Thread(target=load_fixed_job, daemon=True).start()
+
self.active_command_plugins = {}
self.active_service_plugins = {}
self.active_telemetry_plugins = {}
@@ -910,6 +915,37 @@ class SidebandCore():
self.active_service_plugins = {}
self.__load_plugins()
+ def __load_fixed_locations(self):
+ fixed_locations_path = f"{self.config['command_plugins_path']}/fixed_locations"
+ if os.path.isfile(fixed_locations_path):
+ RNS.log(f"Loading fixed locations", RNS.LOG_DEBUG)
+ try:
+ location_entries = []
+ try:
+ with open(fixed_locations_path, "rb") as flf:
+ location_entries = flf.read().decode("utf-8").splitlines()
+ except Exception as e: RNS.log(f"Could not load fixed location entries: {e}", RNS.LOG_ERROR)
+
+ line = 0
+ for e in location_entries:
+ line += 1
+ if e.startswith("#"): continue
+ try:
+ comps = e.replace(" ", "").split(",")
+ if len(comps) == 3:
+ dst = comps[0]; lat = float(comps[1]); lon = float(comps[2])
+ if len(dst) != RNS.Reticulum.TRUNCATED_HASHLENGTH//8*2: raise ValueError(f"Invalid destination hash at line {line}")
+ try: bytes.fromhex(dst)
+ except: raise ValueError(f"Invalid destination hash at line {line}")
+ dst = bytes.fromhex(dst)
+ if lat < -90 or lat > 90: raise ValueError(f"Invalid latitude at line {line}")
+ if lon < -180 or lon > 180: raise ValueError(f"Invalid longitude at line {line}")
+ self.fixed_locations[dst] = {"latitude": lat, "longitude": lon}
+ RNS.log(f"Fixed location set for {RNS.prettyhexrep(dst)}", RNS.LOG_DEBUG)
+ except Exception as e: RNS.log(f"Invalid fixed location entry: {e}", RNS.LOG_ERROR)
+
+ except Exception as e: RNS.log(f"Error while loading fixed locations: {e}", RNS.LOG_ERROR)
+
def __load_plugins(self):
plugins_path = self.config["command_plugins_path"]
command_plugins_enabled = self.config["command_plugins_enabled"] == True
@@ -1777,13 +1813,20 @@ class SidebandCore():
return []
- def owm_location(self):
+ def own_location(self):
return self.peer_location(self.lxmf_destination.hash)
- def peer_location(self, context_dest):
- if context_dest == None:
- return None
+ def fixed_location(self, context_dest):
+ if not context_dest in self.fixed_locations: return None
+ else:
+ try:
+ l = self.fixed_locations[context_dest]
+ if "latitude" in l and "longitude" in l:
+ return {"latitude": float(l["latitude"]), "longitude": float(l["longitude"])}
+ except: return None
+ def peer_location(self, context_dest):
+ if context_dest == None: return None
if context_dest == self.lxmf_destination.hash:
try:
if self.latest_telemetry != None:
@@ -1798,6 +1841,9 @@ class SidebandCore():
except Exception as e:
RNS.log("Error while getting own location: "+str(e), RNS.LOG_ERROR)
+ if self.fixed_location(context_dest):
+ return self.fixed_location(context_dest)
+
after_time = time.time()-3*30*24*60*60
pts = self.peer_telemetry(context_dest, after=after_time)
for pt in pts:
@@ -1808,8 +1854,7 @@ class SidebandCore():
if "latitude" in l and "longitude" in l:
if l["latitude"] != None and l["longitude"] != None:
return l
- except:
- pass
+ except: pass
return None
diff --git a/sbapp/ui/map.py b/sbapp/ui/map.py
index 5533e758..00db840f 100644
--- a/sbapp/ui/map.py
+++ b/sbapp/ui/map.py
@@ -728,6 +728,7 @@ class Map():
latest_viewable = None
peer_interfaces = None
if not skip:
+ fixed_location = self.app.sideband.fixed_location(telemetry_source)
for telemetry_entry in sorted(telemetry_entries[telemetry_source], key=lambda t: t[0], reverse=True):
telemetry_timestamp = telemetry_entry[0]
telemetry_data = telemetry_entry[1]
@@ -736,11 +737,20 @@ class Map():
telemetry = t.read_all()
if "interfaces" in telemetry and telemetry["interfaces"] != None:
if not peer_interfaces: peer_interfaces = telemetry["interfaces"]
+ if fixed_location:
+ latest_viewable = telemetry
+ break
if "location" in telemetry and telemetry["location"] != None and telemetry["location"]["latitude"] != None and telemetry["location"]["longitude"] != None:
latest_viewable = telemetry
break
if latest_viewable != None:
+ if fixed_location:
+ if not "location" in latest_viewable: latest_viewable["location"] = {}
+ latest_viewable["location"]["latitude"] = fixed_location["latitude"]
+ latest_viewable["location"]["longitude"] = fixed_location["longitude"]
+ latest_viewable["location"]["last_update"] = time.time()
+
l = latest_viewable["location"]
if not telemetry_source in self.clustered_markers:
appearance = self.app.sideband.peer_appearance(telemetry_source)
──────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────